package com.example.yunkafka.listener;

import com.example.yunkafka.config.KafkaConfig;
import lombok.extern.slf4j.Slf4j;
import org.apache.kafka.clients.consumer.ConsumerRecord;
import org.apache.kafka.clients.consumer.ConsumerRecords;
import org.springframework.beans.factory.annotation.Autowired;
/**
 * @author CHENTAIWEN
 * @create 2022/8/25 13:26
 */
@Slf4j
public class KafkaListenerJob implements Runnable{

    @Autowired
    private KafkaConsumerListener kafkaConsumerListener;

    //注入消息监听处理类
    public KafkaListenerJob(KafkaConsumerListener kafkaConsumerListener) {
        this.kafkaConsumerListener = kafkaConsumerListener;
    }

    @Override
    public void run() {
        log.info("kafka消息监听任务已启动！");
        //进行消息监听
        while (true) {
            ConsumerRecords<String, String> records = KafkaConfig.kafkaConsumer.poll(1000);
//            已经获取到所有消息
            for (ConsumerRecord<String, String> record : records) {
                try {
                    Thread.sleep(1000);
                    kafkaConsumerListener.listen(record);
                } catch (Exception e) {
                    log.error("消息消费异常！", e);
                }finally {
                    KafkaConfig.kafkaConsumer.commitAsync((offsets, exception) -> {
                        if (exception==null){
                            log.error("手工提交offset成功:"+offsets.toString());
                        }else{
                            log.error("手工提交offset失败:"+offsets.toString());
                            exception.printStackTrace();
                        }
                    });
                }
            }
        }
    }
}
